package com.example.tranport.server.message;

import com.example.destination.JmsDestination;
import com.example.message.JmsMessageWrapper;
import com.example.message.Request;
import com.example.message.RequestType;
import com.example.tranport.server.ServerFactory;
import com.example.utils.RandomUtils;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ServerMessageServiceImpl implements ServerMessageService {

    Logger logger = LoggerFactory.getLogger(ServerMessageServiceImpl.class);

    //存储所有的订阅消息, 保存订阅关系
    private Map<String, List<Long>> destinationAndSessionMap = new ConcurrentHashMap<>();

    //存储所有的订阅关系, 保存订阅关系
    private Map<Long, List<JmsDestination>> sessionAndDestinationMap = new ConcurrentHashMap<>();

    //延迟消息存储的队列
    private LinkedBlockingQueue<Request> requestQueue = new LinkedBlockingQueue<>();

    //服务器工厂类
    private ServerFactory serverFactory;

    //执行线程池，用于处理延迟消息
    private ThreadPoolExecutor threadPoolExecutor;

    //是否运行
    private volatile boolean isRunning = true;

    public ServerMessageServiceImpl(ServerFactory serverFactory) {
        this.serverFactory = serverFactory;
    }

    @Override
    public void start() throws Exception {
        this.threadPoolExecutor = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
        //启动延迟消息处理线程
        this.threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                logger.warn("MQ启动延迟消息队列处理线程 ....");
                while (isRunning) {


                }
            }
        });
    }

    @Override
    public void stop() throws Exception {
        isRunning = false;
        threadPoolExecutor.shutdown();
    }

    @Override
    public void receive(IoSession ioSession, Object o) throws Exception {
        if (o instanceof Request) {
            //注册信息
            if (((Request) o).getType().equalsIgnoreCase(RequestType.REGISTER)) {
                //消费者注册消息的处理
                List<JmsDestination> destinations = (List<JmsDestination>) ((Request) o).getData();
                for (JmsDestination destination : destinations) {
                    if (destinationAndSessionMap.containsKey(destination.getName())) {
                        List<Long> sessionIds = destinationAndSessionMap.get(destination.getName());
                        if (!sessionIds.contains(ioSession.getId())) {
                            sessionIds.add(ioSession.getId());
                            destinationAndSessionMap.put(destination.getName(), sessionIds);
                        }
                    } else {
                        //不包含当前的订阅关系, 则添加订阅消息的
                        List<Long> newSessionIds = new ArrayList<>();
                        newSessionIds.add(ioSession.getId());
                        destinationAndSessionMap.put(destination.getName(), newSessionIds);
                    }
                }
                sessionAndDestinationMap.put(ioSession.getId(), destinations);
            }
            //发送消息队列消息
            if (((Request) o).getType().equalsIgnoreCase(RequestType.SEND_QUEUE)) {
                JmsMessageWrapper jmsMessageWrapper = (JmsMessageWrapper) ((Request) o).getData();
                if (destinationAndSessionMap.containsKey(jmsMessageWrapper.getName())) {
                    List<Long> sessionIds = destinationAndSessionMap.get(jmsMessageWrapper.getName());
                    int index = RandomUtils.random(sessionIds.size());
                    IoSession session = serverFactory.getServerTransPort().getAcceptor().getManagedSessions().get(sessionIds.get(index));
                    if (session != null) {
                        List<JmsDestination> jmsDestinations = sessionAndDestinationMap.get(session.getId());
                        for (JmsDestination jmsDestination : jmsDestinations) {
                            if (jmsDestination.getName().equalsIgnoreCase(jmsMessageWrapper.getName())) {
                                WriteFuture writeFuture = session.write(o);
                                writeFuture.awaitUninterruptibly();
                                if (writeFuture.isWritten()) {
                                    //发送成功
                                } else {
                                    //发送失败
                                }
                            }
                        }

                    }
                }
            }
            //消息的ack确认
            if (((Request) o).getType().equalsIgnoreCase(RequestType.ACK)) {

            }
            serverFactory.getServerStoreService().store((Request) o);
        }
    }

    @Override
    public void closeSession(IoSession ioSession) throws Exception {
        //关闭连接时，移除订阅关系
        for (Map.Entry<String, List<Long>> entry : destinationAndSessionMap.entrySet()) {
            List<Long> sessionIds = entry.getValue();
            if (sessionIds.contains(ioSession.getId())) {
                sessionIds.remove(ioSession.getId());
                if (sessionIds.isEmpty()) {
                    destinationAndSessionMap.remove(entry.getKey());
                } else {
                    destinationAndSessionMap.put(entry.getKey(), sessionIds);
                }
            }
        }
        sessionAndDestinationMap.remove(ioSession.getId());
    }

    @Override
    public void sessionIdle(IoSession ioSession, IdleStatus idleStatus) throws Exception {

    }
}
